/**
 * PriorityBlockingQueue类
 *
 * @author fireway
 * @since 2019年 03月 08日 星期五 06:25:40 CST
 */
package concurrency;

import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;

class PrioritizedTask implements Runnable, Comparable<PrioritizedTask> {
    private final int mPriority;

    protected static List<PrioritizedTask> sSequence = new ArrayList<PrioritizedTask>();

    private Random mRandom = new Random(47);

    private static int sCounter = 0;

    private final int mId = sCounter++;

    public PrioritizedTask(int priority) {
        mPriority = priority;
        sSequence.add(this);
    }

    @Override
    public int compareTo(PrioritizedTask o) {
        return mPriority < o.mPriority ? 1 : (mPriority > o.mPriority ? -1 : 0);
    }

    @Override
    public void run() {
        try {
            TimeUnit.MILLISECONDS.sleep(mRandom.nextInt(250));
        } catch(InterruptedException e) {
            // Acceptable way to exit
        }
        System.out.println(this);
    }

    @Override
    public String toString() {
        return String.format("[%1$-3d]", mPriority) + " Task " + mId;
    }

    public String summary() {
        return "(Task " + mId + ":" + mPriority + ")";
    }

    public static class EndSentinel extends PrioritizedTask {
        private ExecutorService mExecutorService;

        public EndSentinel(ExecutorService e) {
            super(-1);
            mExecutorService = e;
        }

        @Override
        public void run() {
            int count = 0;
            for(PrioritizedTask pt : sSequence) {
                System.out.print(pt.summary() + ", ");
                if(++count % 5 == 0) {
                    System.out.println();
                }

            }
            System.out.println();
            System.out.println(this + " Calling shutdownNow()");
            mExecutorService.shutdownNow();
        }
    }
}

class PrioritizedTaskProducer implements Runnable {
    private Queue<Runnable> mPriorityBlockingQueue;

    private Random mRandom = new Random(47);

    private ExecutorService mExecutorService;

    public PrioritizedTaskProducer(Queue<Runnable> q, ExecutorService e) {
        mPriorityBlockingQueue = q;
        mExecutorService = e;
    }

    @Override
    public void run() {
        // Unbounded queue; never blocks.
        // Fill it up fast with random priorities:
        for(int i = 0; i < 20; i++) {
            mPriorityBlockingQueue.add(new PrioritizedTask(mRandom.nextInt(10)));
            Thread.yield();
        }

        // Trickle in highest-priority jobs:
        try {
            for(int i = 0; i < 10; i++) {
                TimeUnit.MILLISECONDS.sleep(250);
                mPriorityBlockingQueue.add(new PrioritizedTask(10));
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        // Add jobs, lowest priority first:
        for(int i = 0; i < 10; i++) {
            mPriorityBlockingQueue.add(new PrioritizedTask(i));
        }

        // A sentinel to stop all the tasks:
        mPriorityBlockingQueue.add(new PrioritizedTask.EndSentinel(mExecutorService));

        System.out.println("Finished PrioritizedTaskProducer");
    }
}

class PrioritizedTaskConsumer implements Runnable {
    private PriorityBlockingQueue<Runnable> mPriorityBlockingQueue;

    public PrioritizedTaskConsumer(PriorityBlockingQueue<Runnable> q) {
        mPriorityBlockingQueue = q;
    }

    @Override
    public void run() {
        try {
            while(!Thread.interrupted()) {
                // Use current thread to run the task:
                mPriorityBlockingQueue.take().run();
            }
        } catch(InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("Finished PrioritizedTaskConsumer");
    }
}

public class PriorityBlockingQueueDemo {
    public static void main(String[] args) {
        ExecutorService exec = Executors.newCachedThreadPool();
        PriorityBlockingQueue<Runnable> queue = new PriorityBlockingQueue<Runnable>();
        // 注意下面这两句可以调换位置
        exec.execute(new PrioritizedTaskProducer(queue, exec));
        exec.execute(new PrioritizedTaskConsumer(queue));
    }
}
